package com.bw.utils;

import com.bw.bean.TransientSink;
import lombok.SneakyThrows;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class ClickHouseUtil {
    public static <T> SinkFunction<T> sink(String sql) {
        return JdbcSink.sink(sql, new JdbcStatementBuilder<T>() {
                    @SneakyThrows
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                        // 反射拿到所有属性
                        Field[] fields = t.getClass().getDeclaredFields();
                        int offset = 0;
                        for (int i = 0; i < fields.length; i++) {
                            // 为了拿到private
                            fields[i].setAccessible(true);
                            Field field = fields[i];
                            // 拿到属性上的注解
                            TransientSink annotation = field.getAnnotation(TransientSink.class);
                            if (annotation != null) {
                                offset += 1;
                                continue;
                            }
                            // 拿到属性对应的值
                            Object o = field.get(t);
                            preparedStatement.setObject(i + 1 - offset, o);
                        }
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(5)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:clickhouse://hadoop102:8123/default")
                        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                        .build());
    }
}
